package com.cike.sparkstudy.sql.scala

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.functions._

object DailyUV {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("DailyUV")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    import sqlContext.implicits._
    //模拟数据，并把数据转换成RDD
    val userAccessLogs = Array("2019-12-23,1121",
      "2019-12-23,1122",
      "2019-12-23,1121",
      "2019-12-23,1122",
      "2019-12-24,1123",
      "2019-12-24,1122",
      "2019-12-24,1125",
      "2019-12-25,1125")
    val userAccessLogRDDs = sc.parallelize(userAccessLogs,3)

    //将RDD数据转换成DataFrame.
    //  把RDD先转换成RDD<ROW>格式
    val userAccessLogRDDRows = userAccessLogRDDs
      .map{log => Row(log.split(",")(0),log.split(",")(1).toInt)}
    //构造DataFrame的元数据
    val structType = StructType(Array(StructField("date",StringType,true),
      StructField("userid",IntegerType,true)))
    //使用sqlContext创建DataFrame
    val userAccessLogRowDF = sqlContext.createDataFrame(userAccessLogRDDRows,structType)
    userAccessLogRowDF.show()
    //接下来是对DataFrame的操作
    // 这里讲解一下uv的基本含义和业务
    // 每天都有很多用户来访问，但是每个用户可能每天都会访问很多次
    // 所以，uv，指的是，对用户进行去重以后的访问总数

    // 这里，正式开始使用Spark 1.5.x版本提供的最新特性，内置函数，countDistinct
    // 讲解一下聚合函数的用法
    // 首先，对DataFrame调用groupBy()方法，对某一列进行分组
    // 然后，调用agg()方法 ，第一个参数，必须，必须，传入之前在groupBy()方法中出现的字段
    // 第二个参数，传入countDistinct、sum、first等，Spark提供的内置函数
    // 内置函数中，传入的参数，也是用单引号作为前缀的，其他的字段
    userAccessLogRowDF.groupBy("date")
      .agg('date, countDistinct('userid))
      .map { row => Row(row(1), row(2)) }
      .collect()
      .foreach(println)



  }

}
